package com.atguigu.edu.realtime.app.dwd.trade;

import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author maikle
 * @date 2022/8/2
 * 交易域：下单事实表
 * 需要启动的进程
 *      zk、kafka、maxwell、DwdTradeOrderPreProcess、DwdTradeOrderDetail
 */
public class DwdTradeOrderDetail {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(3);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka订单预处理主题中读取数据 创建动态表
        tableEnv.executeSql("" +
                "create table dwd_trade_order_pre_process(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +

                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "od_ts string,\n" +

                "user_id string,\n" +
                "province_id string,\n" +
                "expire_time_date string,\n" +
                "expire_time string,\n" +
                "session_id string,\n" +
                "`type` string,\n" +
                "`old` map<string,string>,\n" +
                "oi_ts string,\n" +

                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));

        //tableEnv.executeSql("select * from dwd_trade_order_pre_process").print();
        //TODO 4.从订单预处理表中过滤出下单行为
        Table filteredTable = tableEnv.sqlQuery("" +
            "select " +
                "id,\n" +
                "order_id,\n" +
                "course_id,\n" +
                "course_name,\n" +
                "date_id,\n" +
                "create_time,\n" +

                "origin_amount,\n" +
                "coupon_reduce,\n" +
                "final_amount,\n" +
                "od_ts,\n" +

                "user_id,\n" +
                "province_id,\n" +
                "expire_time_date,\n" +
                "expire_time,\n" +
                "session_id,\n" +
                "oi_ts,\n" +

                "row_op_ts\n" +
            "from dwd_trade_order_pre_process " +
            "where `type`='insert'");
        tableEnv.createTemporaryView("filtered_table", filteredTable);

        //tableEnv.executeSql("select * from filtered_table").print();
        // TODO 5. 创建 Kafka 下单明细表
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
            "id string,\n" +
            "order_id string,\n" +
            "course_id string,\n" +
            "course_name string,\n" +
            "date_id string,\n" +
            "create_time string,\n" +

            "origin_amount string,\n" +
            "coupon_reduce string,\n" +
            "final_amount string,\n" +
            "od_ts string,\n" +

            "user_id string,\n" +
            "province_id string,\n" +
            "expire_time_date string,\n" +
            "expire_time string,\n" +
            "session_id string,\n" +
            "oi_ts string,\n" +

            "row_op_ts timestamp_ltz(3),\n" +
            "primary key(id) not enforced\n" +
            ")" + KafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail"));

        // TODO 6. 将数据写出到 Kafka
        tableEnv.executeSql("insert into dwd_trade_order_detail select * from filtered_table");

    }
}
